package com;

import com.sink.ASink;
import com.ververica.cdc.connectors.oracle.OracleSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Properties;

/**
 * @description:
 * @author: sksun2
 * @create: 2023-08-14
 */
public class OracleSourceExample {

    public static void main(String[] args) throws Exception {
        Properties debeziumProperties = new Properties();
        debeziumProperties.setProperty("log.mining.strategy", "online_catalog");
        debeziumProperties.setProperty("log.mining.continuous.mine", "true");
        debeziumProperties.setProperty("debezium.database.tablename.case.insensitive", "false");


        SourceFunction<String> sourceFunction = OracleSource.<String>builder()
//                .url("jdbc:oracle:thin:@{hostname}:{port}:{database}")
                .url("jdbc:oracle:thin:@10.7.204.250:1521:pomppre")
                .hostname("10.7.204.250")
                .port(1521)
                .database("pomppre") // monitor XE database
                .schemaList("pomp") // monitor inventory schema
//                .tableList("pomp.PROJECT_BASE_INFO") // monitor products table
                .tableList("pomp.PROJECT_TYPE_INFO") // monitor products table
                .username("pomp")
                .password("pomp#pre##")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
                .addSource(sourceFunction)
//                .addSink(new CustomSink())
                .addSink(new ASink())
//                .print()
                .setParallelism(2);
               // .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering

        env.execute();
    }

}
